package com.cctc.rds.scheduler.server.schedule;

import com.cctc.rds.nrpc.core.util.JsonUtils;
import com.cctc.rds.scheduler.core.task.TaskLog;
import com.cctc.rds.scheduler.core.task.TaskStatus;
import com.cctc.rds.scheduler.server.raft.helper.RaftHelper;
import com.cctc.rds.scheduler.server.schedule.hashedwheel.HashedWheelTimer;
import com.cctc.rds.scheduler.server.schedule.hashedwheel.TimerTask;
import com.cctc.rds.scheduler.server.schedule.hashedwheel.TimerTaskCallback;
import com.cctc.rds.scheduler.server.storage.Storage;
import com.cctc.rds.scheduler.server.storage.impl.LevelDBStorage;
import com.cctc.rds.scheduler.server.storage.leveldb.LevelDB;
import com.cctc.rds.scheduler.server.storage.leveldb.manager.ClearHistoryTaskWorker;
import com.cctc.rds.scheduler.server.storage.leveldb.manager.RecoveryLockTaskWorker;
import org.iq80.leveldb.DB;
import org.iq80.leveldb.DBIterator;
import org.iq80.leveldb.Snapshot;

import java.io.Closeable;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 任务调度服务，[服务端] 实现类。
 * 任务 [提交 -> 调度 -> 执行] 流程：
 * Step 1：任务调度器使用方通过任务调度器客户端提供任务类型或任务实例，并将其放入到任务实例工厂内，然后指定任务延迟执行周期，向 Leader 节点提交任务实例。
 * 其中，任务实例会被封装成 [任务日志（penging）] 同步至所有节点，存储到本机 LevelDB；
 * Step 2：Leader 节点扫描本机存储状态为 pending 的任务实例日志放入时间轮（阻塞队列、待调度），更新任务状态为 scheduling；使用时间轮算法，自动将到期
 * 执行的任务实例调度到执行容器内（已调度，待执行）；
 * Step 3：任务调度器客户端自动从 Leader 节点的执行容器中拉取任务并执行，执行完成后向 Leader 节点更新任务执行结果（任务状态：success、fail），并将
 * 更新后的任务实例日志（任务状态：success、fail）同步至所有 Follower 节点；
 * --> 如果任务实例日志状态为 success（执行成功），则结束执行。
 * --> 如果任务实例日志状态为 fail（执行失败），则设置重新调度时间为当前时间，并计算下一次延迟执行时间，更新任务状态为 pending。
 * Step 4：任务调度器使用方通过任务调度器客户端取消任务实例，终止调度执行。
 * 此外，对于 Raft 集群：
 * - 当 Leader 节点挂掉时，对于已经提交到任务调度器服务端的任务，或者没有任务调度器客户端拉取，或者已经被任务调度器客户端拉取但未提交执行结果的任务，都会被
 * 新的 Leader 节点锁定一段时间，然后重置为 pending 状态，重新调度，延迟执行。因此，任务调度器使用方需要自行确保任务实例的幂等执行。
 */
public class TaskScheduleServer implements Runnable, TimerTaskCallback, Closeable {
    private Storage storage;
    private ClearHistoryTaskWorker clearHistoryTaskWorker;
    private RecoveryLockTaskWorker recoveryLockTaskWorker;
    private HashedWheelTimer hashedWheelTimer;
    private Map<String, Queue<Long>> container;
    private Thread scheduler;
    private final AtomicBoolean status;

    public TaskScheduleServer() {
        this.storage = new LevelDBStorage();
        this.clearHistoryTaskWorker = new ClearHistoryTaskWorker();
        this.recoveryLockTaskWorker = new RecoveryLockTaskWorker();
        this.hashedWheelTimer = new HashedWheelTimer(this);
        this.container = new HashMap<>();
        this.scheduler = new Thread(this, "delay-schedule-server");
        this.status = new AtomicBoolean(Boolean.TRUE);
        this.clearHistoryTaskWorker.start();
        this.recoveryLockTaskWorker.start();
        this.scheduler.start();
    }

    public boolean offer(String app, long taskId, TaskLog taskLog) {
        long scheduleTime = taskLog.getRetryScheduleTime() > 0 ?
                taskLog.getRetryScheduleTime() : taskLog.getScheduleTime();
        long execTime = taskLog.getExecPriod().get(taskLog.getExecPriodIndex());
        long deadline = scheduleTime + execTime;
        if (deadline - System.currentTimeMillis() / 1000 > 60) return false;
        String taskKey = app + "::" + taskId;
        hashedWheelTimer.addTask(taskKey, scheduleTime, execTime);
        return true;
    }

    public List<Long> pull(String app, int maxRecord) {
        Queue<Long> queue;
        synchronized (this) {
            queue = container.get(app);
        }
        List<Long> ans = new ArrayList<>(Math.min(maxRecord, queue.size()));
        for (int i = 0; i < ans.size(); i++) {
            Long taskId = queue.poll();
            if (taskId == null) break;
            ans.add(taskId);
        }
        return ans;
    }

    @Override
    public void run() {
        out:
        while (!Thread.interrupted()) {
            try {
                RaftHelper.checkLeader();
                status.set(Boolean.TRUE);
            } catch (Throwable e) {
                try {
                    TimeUnit.MILLISECONDS.sleep(100);
                } catch (InterruptedException ex) {
                } finally {
                    continue;
                }
            }
            while (status.get()) {
                DB database = LevelDB.get();
                if (database == null) break out;
                try (Snapshot snapshot = database.getSnapshot(); DBIterator iterator = database.iterator()) {
                    String key = TaskStatus.PENDING.getStatus() + "::";
                    iterator.seek(key.getBytes(StandardCharsets.UTF_8));
                    while (iterator.hasNext()) {
                        Map.Entry<byte[], byte[]> entry = iterator.next();
                        String[] taskInfo = new String(entry.getKey(), StandardCharsets.UTF_8).split("::");
                        String app = taskInfo[1];
                        long taskId = Long.parseLong(taskInfo[2]);
                        String value = new String(entry.getValue(), StandardCharsets.UTF_8);
                        TaskLog taskLog = JsonUtils.fromJson(value, TaskLog.class);
                        if (!taskLog.getStatus().equals(TaskStatus.PENDING.getStatus())) break;
                        if (offer(app, taskId, taskLog)) {
                            taskLog.setStatus(TaskStatus.SCHEDULING.getStatus());
                            storage.save(app, taskId, taskLog);
                        }
                    }
                } catch (IOException e) {
                    break out;
                }
            }
            try {
                RaftHelper.checkLeader();
                TimeUnit.MILLISECONDS.sleep(10);
            } catch (InterruptedException e) {
                status.set(Boolean.FALSE);
            }
        }
    }

    @Override
    public void exec(List<TimerTask> tasks) {
        try {
            RaftHelper.checkLeader();
        } catch (Throwable e) {
            synchronized (this) {
                container.forEach((key, value) -> {
                    value.clear();
                });
                container.clear();
            }
        }
        tasks.forEach(task -> {
            String[] taskInfo = task.getTaskKey().split("::");
            String app = taskInfo[0];
            long taskId = Long.parseLong(taskInfo[1]);
            synchronized (this) {
                container.computeIfAbsent(app, i -> new LinkedList<>()).add(taskId);
            }
        });
    }

    @Override
    public void close() throws IOException {
        status.set(Boolean.FALSE);
        clearHistoryTaskWorker.close();
        recoveryLockTaskWorker.close();
        hashedWheelTimer.close();
        scheduler.interrupt();
    }

}
